package util;

import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.List;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.KeyRange;
import org.apache.cassandra.thrift.KeySlice;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;

import me.prettyprint.cassandra.model.BasicColumnDefinition;
import me.prettyprint.cassandra.model.BasicColumnFamilyDefinition;
import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.cassandra.service.CassandraHostConfigurator;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.ddl.ColumnFamilyDefinition;
import me.prettyprint.hector.api.ddl.ColumnIndexType;
import me.prettyprint.hector.api.ddl.ComparatorType;
import me.prettyprint.hector.api.ddl.KeyspaceDefinition;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;

public class CassandraService {

	// The string serializer translates the byte[] to and from String using
	// utf-8 encoding
	private static StringSerializer stringSerializer = StringSerializer.get();

	private Cluster cluster;
	private String clusterName = "TestCluster";
	private String clusterHost = "localhost:9160";
	private String keyspaceName = "Crawl";
	private KeyspaceDefinition keyspaceDef;
	private Keyspace keyspace;
	private ColumnFamilyDefinition cfDef;
	private Mutator<String> mutator;
	
	public int returnedCounter = 0;

	public CassandraService() {
		/*
		 * 1. Connect to cluster 
		 * 2. Define column family
		 */
		
		try {
			// Create a cluster object
			cluster = HFactory.getOrCreateCluster(clusterName,
					new CassandraHostConfigurator(clusterHost));

			keyspaceDef = cluster.describeKeyspace(keyspaceName);
			if (keyspaceDef == null) {
				keyspaceDef = HFactory.createKeyspaceDefinition(keyspaceName);
				cluster.addKeyspace(keyspaceDef);
			} else {
				System.out.println("keyspace " + "\"" + keyspaceName + "\""
						+ " already exist!");
			}

			// keyspace object, doesn't create the actual keyspace, just a
			// reference for usage
			keyspace = HFactory.createKeyspace(keyspaceName, cluster);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public void setupColumnFamily(String columnFamily) {
		// create UTF8Type column
		// add schema to column
		// set key type to utf8, (assume "column" keys as utf8)
		List<ColumnFamilyDefinition> cfDefList = keyspaceDef.getCfDefs();
		if (cfDefList.size() == 0) {
			// no columnfamily, make new
			cfDef = HFactory.createColumnFamilyDefinition(keyspaceName, columnFamily,
					ComparatorType.UTF8TYPE);
			cluster.addColumnFamily(cfDef);
		} else if (cfDefList.size() > 0) {
			// has columnfamily, check what it is
			String thisCfName = cfDefList.get(0).getName();
			if (!thisCfName.equals(columnFamily)) {
				cfDef = HFactory.createColumnFamilyDefinition(keyspaceName,
						columnFamily, ComparatorType.UTF8TYPE);
				cluster.addColumnFamily(cfDef);
			} else {
				System.out.println("columnFamily " + "\"" + columnFamily + "\""
						+ " already exist!");
			}
		}
		/*
        BasicColumnDefinition columnDefinition = new BasicColumnDefinition();
        columnDefinition.setName(stringSerializer.toByteBuffer("html_doc"));
        columnDefinition.setIndexName("html_doc_idx");
        columnDefinition.setIndexType(ColumnIndexType.KEYS);
        columnDefinition.setValidationClass(ComparatorType.UTF8TYPE.getClassName());
        
        BasicColumnFamilyDefinition columnFamilyDefinition = new BasicColumnFamilyDefinition();
        columnFamilyDefinition.setKeyspaceName(keyspaceName);
        columnFamilyDefinition.setName(columnFamily);
        columnFamilyDefinition.addColumnDefinition(columnDefinition);
        */	
	}

	public void insertHtmlData(String columnFamily, String rowKey, String key,
			String value) {
		// Create a mutator object for this keyspace using utf-8 encoding
		mutator = HFactory.createMutator(keyspace, stringSerializer);
		mutator.insert(rowKey, columnFamily,
				HFactory.createStringColumn(key, value));
	}
	
	public HashMap<String,Integer> queryHtmlData(String columnFamily, int numRowKeys, String rowKey, String key) throws InvalidRequestException, TException, UnavailableException, TimedOutException {
		/*
		 * connection and query setup
		 */
		TTransport tr = new TFramedTransport(new TSocket("localhost", 9160));
		TProtocol proto = new TBinaryProtocol(tr);

		Cassandra.Client client = new Cassandra.Client(proto);
		tr.open();
		client.set_keyspace(keyspaceName);
		
		ColumnParent columnParent = new ColumnParent(columnFamily);

		SlicePredicate predicate = new SlicePredicate();
		predicate.setSlice_range(new SliceRange(ByteBuffer.wrap(new byte[0]),
				ByteBuffer.wrap(new byte[0]), false, 5000));

		KeyRange keyRange = new KeyRange(5000);
		keyRange.setStart_key(new byte[0]);
		keyRange.setEnd_key(new byte[0]);
		List<KeySlice> keySlices = client.get_range_slices(columnParent,
				predicate, keyRange, ConsistencyLevel.ONE);
		
		Charset UTF8_CHARSET = Charset.forName("UTF-8");
		
		// HashMap<url, count>
		HashMap<String, Integer> unSorted = new HashMap<String, Integer>();
	
		/*
		 * for all the rowKeys (i)
		 * 		count the total columns in that rowKey
		 * 			for all the columns in that rowKey (j)
		 * 				get the first <k,v>
		 * 				compare <k> to user search word
		 * 				if match
		 * 					store the <v> and search for the html_url
		 */
		for(int i = 0; i < numRowKeys; i++) { // numRowKeys = 76 for sjsu site
			int columnsCount = keySlices.get(i).getColumns().size(); // how many key/value pairs in the rowKey (html_x)
			System.out.println("i=" + i + " : " + "columnsCount = " + columnsCount);
			String savedUrl = null;
			String currentKey = null;
			String currentKeyValue = null;
			for(int j = 0; j < columnsCount; j++) { // for all the key/value pairs in this rowKey
				// get the name of the key of <key,value> 
				currentKey = new String(keySlices.get(i).getColumns().get(j).getColumn().getName(), UTF8_CHARSET);
				// can't assign as Integer if the currentKey is either html_url or html_content
				currentKeyValue = new String(keySlices.get(i).getColumns().get(j).getColumn().getValue(), UTF8_CHARSET);
				
				if(currentKey.equals(key)) {
					// debug
					System.out.println("i=" + i + " : " + "j=" + j + " : " + "key=" + currentKey + "," + "value=" + currentKeyValue);
					
					// loop through again to find htm_url
					for(int k = 0; k < columnsCount; k++) {
						String htmlKeySearch = new String(keySlices.get(i).getColumns().get(k).getColumn().getName(), UTF8_CHARSET);
						if(htmlKeySearch.equals("html_url")) {
							String htmlValueSearch = new String(keySlices.get(i).getColumns().get(k).getColumn().getValue(), UTF8_CHARSET);
							savedUrl = htmlValueSearch;
						}
					}
					
					// put(url, count> <string, Integer>
					unSorted.put(savedUrl, Integer.parseInt(new String(keySlices.get(i).getColumns().get(j).getColumn().getValue(), UTF8_CHARSET)));
					
					// debug
		            System.out.println("The search word is: " + key);
		            System.out.println("If this prints, that means key was found and has value of : " + currentKey);
		            System.out.println(new String(rowKey + i));
					System.out.println(currentKey + " appears " + new String(keySlices.get(i).getColumns().get(j).getColumn().getValue(), UTF8_CHARSET) + " times in " + savedUrl);
					System.out.println();
				}
			}
		}
			
		tr.close();
		
		return unSorted;
	}
	
	public void shutDown() {
		cluster.getConnectionManager().shutdown();
	}
}
